package org.apache.rocketmq.namesrv.processor;

import com.alibaba.fastjson.serializer.SerializerFeature;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetRouteInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/27
 * @方法描述：专门处理生产者、消费者客户端发送的根据主题获取路由信息请求的处理器
 */
public class ClientRequestProcessor implements NettyRequestProcessor {


    private static Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    protected NamesrvController namesrvController;

    //Namesrv模块开始启动的时间
    private long startupTimeMillis;

    //这个标志是用来表示是否还需要检查Namesrv准备好提供服务了吗
    //如果为false，就意味着不必检查，Namesrv可以直接对外提供服务了
    //如果为true，则意味着还需要检查，那么Namesrv就不能够对外提供服务
    private AtomicBoolean needCheckNamesrvReady = new AtomicBoolean(true);

    //构造方法
    public ClientRequestProcessor(final NamesrvController namesrvController) {
        this.namesrvController = namesrvController;
        this.startupTimeMillis = System.currentTimeMillis();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：处理客户端请求的方法
     */
    @Override
    public RemotingCommand processRequest(final ChannelHandlerContext ctx, final RemotingCommand request) throws Exception {
        return this.getRouteInfoByTopic(ctx, request);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：根据主题获取对应路由信息的方法，在第二版本代码中，该方法并没有得到实现，是一个伪方法，仅仅是返回客户端一行字符串而已，等引入了broker，我就会为大家真正实现这个方法了
     */
    public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        //创建一个Remoting协议对象，用来封装返回给客户端的响应
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        //获得请求头，本来后面要执行的操作就是根据请求头中的主题信息获取对应的路由信息，但现在还无法实现这个操作，所以我就把操作都省略了
        final GetRouteInfoRequestHeader requestHeader = (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);

        //这里有一个比较重要的操作，那就是判断当前Namesrv是否启动完毕，也就是说是否可以完全对外提供服务了。namesrvController.getNamesrvConfig().getWaitSecondsForService()得到的是配置的Namesrv启动之后应该等待的时间
        //这段时间用来给Namesrv启动内部各个组件，然后判断当前时间减去系统启动时间是否大于等待时间了，如果大于等待时间，就意味着Namesrv可以对外提供服务了
        boolean namesrvReady = needCheckNamesrvReady.get() && System.currentTimeMillis() - startupTimeMillis >= TimeUnit.SECONDS.toMillis(namesrvController.getNamesrvConfig().getWaitSecondsForService());
        if (namesrvController.getNamesrvConfig().isNeedWaitForService() && !namesrvReady) {
            //执行到这里就意味着还没等待足够的时间，那么就要继续等待一会才能让Namesrv对外提供服务
            log.warn("name server not ready. request code {} ", request.getCode());
            //直接回复客户端“name server not ready”消息即可
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("name server not ready");
            return response;
        }
//程序执行到这里意味着Namesrv已经准备好了，可以对外提供服务了，这个时候就可以从Namesrv的路由信息管理器中获取主题对应的路由信息了
        TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

        if (topicRouteData != null) {
            //如果得到了路由信息，就把needCheckNamesrvReady设置为false，表示不需要再检查Namesrv是否准备好了，因为已经准备好了
            if (needCheckNamesrvReady.get()) {
                needCheckNamesrvReady.set(false);
            }
            //下面这段代码暂时注释了，目前还用不到
//            if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
//                String orderTopicConf = this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG, requestHeader.getTopic());
//                topicRouteData.setOrderTopicConf(orderTopicConf);
//            }

            //定义一个字节数组，用来存放编码后的路由信息
            byte[] content;
            //从请求头中判断是否只接受标准JSON格式的响应数据
            Boolean standardJsonOnly = Optional.ofNullable(requestHeader.getAcceptStandardJsonOnly()).orElse(false);
            //如果当前框架版本大于等于4.9.4，或者请求头中标识只接受标准JSON格式的响应数据，那么就使用指定的序列化特性进行编码
            if (request.getVersion() >= MQVersion.Version.V4_9_4.ordinal() || standardJsonOnly) {
                //在这里对获得到的路由信息进行编码，编码后的内容存放到content中
                content = topicRouteData.encode(SerializerFeature.BrowserCompatible,
                        SerializerFeature.QuoteFieldNames, SerializerFeature.SkipTransientField,
                        SerializerFeature.MapSortField);
            } else {
                //这里就是使用默认的序列化特性进行编码
                content = topicRouteData.encode();
            }

            //设置响应体到响应对象中
            response.setBody(content);
            //设置成功响应码
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
            return response;
        }
        //程序执行到这里意味着没有从路由信息管理器中获取到主题对应的路由信息，这时候就要返回主题不存在的响应给客户端
        response.setCode(ResponseCode.TOPIC_NOT_EXIST);
        //设置备注信息，告诉客户端没有找到对应的主题路由信息
        response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic() + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
        return response;
    }


    @Override
    public boolean rejectRequest() {
        return false;
    }
}
